package com.bw.gmall.realtime.common.Util;

import com.bw.gmall.realtime.common.constant.Constant;
import com.sun.org.apache.xalan.internal.xsltc.compiler.util.Type;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

public class FlinkSourceUtil {


    //消费kafka

    public static KafkaSource<String> getKafkaSource(String topic,String groupid){

        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setGroupId(groupid)
                .setTopics(topic)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] bytes) throws IOException {

                        if(bytes != null){
                            return new String(bytes, StandardCharsets.UTF_8);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return Types.STRING;
                    }
                })
                .build();
    }


//        读取CDC
        public static MySqlSource<String> getMysqlSource(String DataBase,String tableName){


//        配置项
            Properties properties = new Properties();
            properties.setProperty("useSSL","false");

            //AllowPublicKeyRetrieval=True可能会导致恶意的代理通过中间人攻击(MITM)获取到明文密码，
            // 所以默认是关闭的，必须显式开启
            properties.setProperty("allowPublicKeyRetrieval","true");

            return MySqlSource.<String>builder()
                    .hostname(Constant.MYSQL_HOST)
                    .port(Constant.MYSQL_PORT)
                    .username(Constant.MYSQL_USER_NAME)
                    .password(Constant.MYSQL_USER_PASSWORD)
                    .databaseList(DataBase)
                    // 库名.表名
                    .tableList(DataBase+"."+tableName)
                    .jdbcProperties(properties)
                    // 序列化器
                    .deserializer(new JsonDebeziumDeserializationSchema())
                    // 初始化
                    .startupOptions(StartupOptions.initial())
                    .build();



        };


}
